package com.ehl.capture.data;

import com.ehl.capture.bean.CarPlaBean;
import com.ehl.capture.db.Config;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.message.MessageAndMetadata;
import org.apache.log4j.Logger;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class WorkThread implements Runnable {
    private IPassCar passcar;
    private KafkaStream<byte[], byte[]> stream;
    private static Logger logger = Logger.getLogger(WorkThread.class);
    private ExecutorService executor = Executors.newFixedThreadPool(Integer.parseInt(Config.getValue("threadcount")));

    public WorkThread(IPassCar passcar, KafkaStream<byte[], byte[]> stream) {
        this.stream = stream;
        this.passcar = passcar;
    }

    public void run() {
        ConsumerIterator it = this.stream.iterator();
        while(it.hasNext()) {
            try {
                MessageAndMetadata<byte[], byte[]> message = it.next();
                long offset = it.kafka$consumer$ConsumerIterator$$consumedOffset();
                int partition = it.kafka$consumer$ConsumerIterator$$currentTopicInfo().partitionId();
                ByteBuffer buf = ByteBuffer.wrap(message.message());
                buf.order(ByteOrder.LITTLE_ENDIAN);
                int dataLength = buf.getInt();
                byte[] dataByte = new byte[buf.remaining()];
                buf.get(dataByte, 0, dataLength);
                String content = new String(dataByte, "UTF-8");
                //String content = new String (message.message(),"utf-8");

                final CarPlaBean bean = this.passcar.strToBean(content);
                if (bean.getPasstime() != null && !bean.getPasstime().equals("null") && !bean.getPasstime().equals("")
                        && bean.getTgsid() != null && !bean.getTgsid().equals("null") && !bean.getTgsid().equals("")) {
                    logger.info("接收过车数据：" + content.replaceAll("\n", "~"));
                    this.passcar.setUrl(bean, "" + partition, offset);
                    this.executor.submit(new Runnable() {
                        public void run() {
                            WorkThread.this.passcar.handData(bean);
                        }
                    });
                }
            } catch (Exception var11) {
                logger.error("读取kafka异常！", var11);
                var11.printStackTrace();
            }
        }

    }

    public static byte[][] getMoreImage(byte[] message) {
        int len = getInt(message, 0);
        byte imgCount = message[4 + len];
        int curLen = 4 + len + 1;
        byte[][] images = new byte[imgCount][];

        for(byte i = 0; i < imgCount; ++i) {
            len = getInt(message, curLen);
            curLen += 4;
            images[i] = getBytes(message, curLen, len);
            curLen += len;
        }

        return images;
    }

    private static int getInt(byte[] bb, int index) {
        return (bb[index + 3] & 255) << 24 | (bb[index + 2] & 255) << 16 | (bb[index + 1] & 255) << 8 | (bb[index + 0] & 255) << 0;
    }

    private static byte[] getBytes(byte[] b, int start, int length) {
        byte[] res = new byte[length];
        System.arraycopy(b, start, res, 0, length);
        return res;
    }
}
